package com.guli.secKill.service.impl;
/*
 * @Author 罗俊
 * @date 2020/12/1 - 2:19 下午
 *
 */

import com.alibaba.csp.sentinel.Entry;
import com.alibaba.csp.sentinel.SphU;
import com.alibaba.csp.sentinel.annotation.SentinelResource;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.guli.common.constant.OrderConstant;
import com.guli.common.constant.SecKillConstant;
import com.guli.common.pojo.SeckillSessionEntity;
import com.guli.common.pojo.SeckillSkuRelationEntity;
import com.guli.common.pojo.vo.SkuVo;
import com.guli.common.pojo.vo.StatusCodeEnum;
import com.guli.common.to.MemberTo;
import com.guli.common.to.SecKillOrderInfoTo;
import com.guli.common.to.SecKillSessionAndRelation;
import com.guli.mall.serviceBase.globalException.MallException;
import com.guli.secKill.interceptor.LoginInterceptor;
import com.guli.secKill.schedule.SecKillSkuScheduled;
import com.guli.secKill.service.SecKillService;
import com.guli.secKill.service.impl.blockHandler.SecKillServiceBlockHandler;
import com.guli.secKill.service.impl.fallBack.SecKillServiceFallBack;
import com.guli.secKill.vo.SecKillInfoVo;
import com.guli.secKill.vo.SecKillRelationAndSpu;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import java.util.stream.Collectors;


@Service
@Slf4j
public class SecKillServiceImpl implements SecKillService {

    @Resource
    RedisTemplate redisTemplate;

    @Resource
    SecKillSkuScheduled secKillSkuScheduled;

    @Resource
    RedissonClient redissonClient;

    @Resource
    RabbitTemplate rabbitTemplate;

    @SentinelResource(value = "getAvailableSecKillSpuResource",
            blockHandlerClass = SecKillServiceBlockHandler.class,
            blockHandler = "getAvailableSecKillSpuBlockHandler",
            fallbackClass = SecKillServiceFallBack.class,
            fallback = "getAvailableSecKillSpu"
    )
    @Override
    public List<SecKillRelationAndSpu> getAvailableSecKillSpu() {

        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(SecKillConstant.UPLOAD_SKU_LOCK);

        RLock rLock = readWriteLock.readLock();

        rLock.lock();

        try {

            BoundHashOperations<String, String, List<SeckillSkuRelationEntity>> hashOps = redisTemplate.boundHashOps(SecKillConstant.PROMOTION_SESSION_CACHE_PREFIX);

            BoundHashOperations<String, String, SecKillRelationAndSpu> boundHashOps = redisTemplate.boundHashOps(SecKillConstant.PRODUCT_SPU_PREFIX);

//        ValueOperations<String, SecKillRelationAndSpu> opsForValue = redisTemplate.opsForValue();

            Long nowTime = new Date().getTime();

            List<SecKillRelationAndSpu> spus = new ArrayList<>();

            hashOps.keys().stream().filter(key -> {

                String[] s = key.split("_");

                long startTime = Long.parseLong(s[0]);

                long endTime = Long.parseLong(s[1]);

                if (nowTime >= startTime && nowTime <= endTime) {
                    return true;
                }
                return false;
            }).forEach(key -> {

                List<SeckillSkuRelationEntity> relations = hashOps.get(key);

                List<SecKillRelationAndSpu> collect = relations.stream().map(relation -> {

                    return boundHashOps.get(relation.getPromotionSessionId() + "_" + relation.getSpuId());

//                secKillRelationAndSpu.setRandomCode(null);

//                return secKillRelationAndSpu;

                }).filter(secKillRelationAndSpu -> {
                    return !ObjectUtils.isEmpty(secKillRelationAndSpu);
                }).collect(Collectors.toList());

                if (!CollectionUtils.isEmpty(collect)) {

                    spus.addAll(collect);
                }

            });
            return spus;
        } finally {
            rLock.unlock();
        }
    }

    @Override
    public List<SecKillRelationAndSpu> checkWhetherTheProductParticipatesInTheSpike(String spuId) {


        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(SecKillConstant.UPLOAD_SKU_LOCK);

        RLock rLock = readWriteLock.readLock();

        rLock.lock();
        try {
            BoundHashOperations<String, String, SecKillRelationAndSpu> boundHashOps = redisTemplate.boundHashOps(SecKillConstant.PRODUCT_SPU_PREFIX);

            List<String> collect = boundHashOps.keys().stream().filter(key -> {
                return Pattern.matches("^\\d" + "_" + spuId, key);
            }).collect(Collectors.toList());

            if (CollectionUtils.isEmpty(collect)) {
                return null;
            }

            List<SecKillRelationAndSpu> secKillRelationAndSpus = boundHashOps.multiGet(collect);

            long nowTime = System.currentTimeMillis();

            // 过滤掉秒杀活动已经结束了的商品
            return secKillRelationAndSpus.stream().filter(relationAndSpu -> {

                if (relationAndSpu.getStartTime() > nowTime) {
                    // 秒杀活动还没有开始，不发送秒杀随机码
                    for (SkuVo skuVo : relationAndSpu.getSpuAndSkuListTo().getSkuVos()) {
                        skuVo.setRandomCode(null);
                    }
                }


                return relationAndSpu.getEndTime() > nowTime;
            }).collect(Collectors.toList());

        } finally {
            rLock.unlock();
        }
    }

    @Override
    public List<SecKillSessionAndRelation> checkWhetherTheProductParticipatesInTheSpikeAndReturnRelation(String spuId) {

        List<SecKillRelationAndSpu> secKillRelationAndSpus = null;
        // 1.5.0 版本开始可以利用 try-with-resources 特性（使用有限制）
        // 资源名可使用任意有业务语义的字符串，比如方法名、接口名或其它可唯一标识的字符串。
        // 自定义自原名进行限流
        try (Entry entry = SphU.entry("checkWhetherTheProductParticipatesInTheSpike")) {
            // 被保护的业务逻辑
            // do something here...

            secKillRelationAndSpus = this.checkWhetherTheProductParticipatesInTheSpike(spuId);
        } catch (BlockException ex) {
            // 资源访问阻止，被限流或被降级
            // 在此处进行相应的处理操作
            log.error("流量过大，资源被限流：{}", ex.getMessage());
        }


        if (CollectionUtils.isEmpty(secKillRelationAndSpus)) {
            return null;
        }

        return secKillRelationAndSpus.stream().map(relationAndSpu -> {

            SecKillSessionAndRelation secKillSessionAndRelation = new SecKillSessionAndRelation();

            secKillSessionAndRelation.setSeckillSkuRelationEntity(relationAndSpu.getSeckillSkuRelation());

            SeckillSessionEntity seckillSessionEntity = new SeckillSessionEntity();

            seckillSessionEntity.setStartTime(new Date(relationAndSpu.getStartTime()));

            seckillSessionEntity.setEndTime(new Date(relationAndSpu.getEndTime()));

            secKillSessionAndRelation.setSeckillSessionEntity(seckillSessionEntity);

            //还是要把随机码返回给product微服务的
            secKillSessionAndRelation.setSkuVos(relationAndSpu.getSpuAndSkuListTo().getSkuVos());
//            secKillSessionAndRelation.setRandomCode(relationAndSpu.getRandomCode());

            return secKillSessionAndRelation;

        }).sorted((secKillSessionAndRelation1, secKillSessionAndRelation2) -> {
            // 按照秒杀活动的开始时间进行排序
            return (int) (secKillSessionAndRelation1.getSeckillSessionEntity().getStartTime().getTime() - secKillSessionAndRelation2.getSeckillSessionEntity().getStartTime().getTime());
        }).collect(Collectors.toList());

    }

    @Override
    public void refreshCache() {

        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(SecKillConstant.UPLOAD_SKU_LOCK);

        RLock rLock = readWriteLock.writeLock();
        rLock.lock();
        try {


            redisTemplate.delete(SecKillConstant.PRODUCT_SPU_PREFIX);

            redisTemplate.delete(SecKillConstant.PROMOTION_SESSION_CACHE_PREFIX);

        } finally {
            rLock.unlock();
        }

//        redisTemplate.delete(SecKillConstant.REDIS_SON_SPU_SEMAPHORE_PREFIX);

        //TODO 秒杀商品的库存（即信号量）尚未删除，因为信号量和秒杀商品服务没有在同一个redis库，删除信号量时，要先获取秒杀商品的随机码
//        redissonClient.
        log.info("秒杀的商品缓存已经删除，现执行写入缓存操作");

        secKillSkuScheduled.uploadSkuToRedis();
    }

    @Override
    public String executeSecKill(SecKillInfoVo secKillInfoVo) {

        // 1、获取秒杀商品的详细信息
        BoundHashOperations<String, String, SecKillRelationAndSpu> relationAndSpuMapOps = redisTemplate.boundHashOps(SecKillConstant.PRODUCT_SPU_PREFIX);


        SecKillRelationAndSpu secKillRelationAndSpu = relationAndSpuMapOps.get(secKillInfoVo.getPromotionSessionId() + "_" + secKillInfoVo.getSpuId());

//        BoundHashOperations<String, String, Integer> userPurchaseFlagOps = redisTemplate.boundHashOps(SecKillConstant.USER_SEC_KILL_FLAG);
        ValueOperations<String, Integer> userPurchaseFlagOps = redisTemplate.opsForValue();


        // 2、校验秒杀信息的合法性
        long nowTime = System.currentTimeMillis();

        // 2.1 校验秒杀活动是否开始或是否结束
        if (nowTime < secKillRelationAndSpu.getStartTime() || nowTime > secKillRelationAndSpu.getEndTime()) {
            throw new MallException(StatusCodeEnum.SEC_KILL_NOT_START_YET_OR_HAS_BEEN_TERMINATED);
        }

        // 2.2 校验商品的随机码和商品id(spuId)是否匹配
        SecKillOrderInfoTo secKillOrderInfoTo = new SecKillOrderInfoTo();

        List<SkuVo> collect = secKillRelationAndSpu.getSpuAndSkuListTo().getSkuVos().stream().filter(skuVo -> {

            secKillOrderInfoTo.setPrice(skuVo.getPrice()); // 找到了就设置价格

            return secKillInfoVo.getSpuId().equals(skuVo.getSpuId()) &&
                    secKillInfoVo.getSkuId().equals(skuVo.getSkuId()) &&
                    secKillInfoVo.getRandomCode().equals(skuVo.getRandomCode());
        }).collect(Collectors.toList());

        if (CollectionUtils.isEmpty(collect)) {
            throw new MallException(StatusCodeEnum.SEC_KILL_SPU_ID_NOT_MATCH_RANDOM_CODE);
        }

        // 2.3 验证用户单次的购买数量是否合理
        if (secKillInfoVo.getNumber() > secKillRelationAndSpu.getSeckillSkuRelation().getSeckillLimit()) {
            throw new MallException(StatusCodeEnum.SEC_KILL_PURCHASE_QUANTITY_EXCEEDS_LIMIT);
        }

        /* 2.4 验证本用户之前有没有购买过此商品（幂等性），如果购买过，那么购买过的数量+此次要购买的数量大于单个用户可购买的数量，那么就抛异常
                只要秒杀成功，就去 redis 中占位
              通过使用 userId_sessionId_spuId 作为 key，标示这个用户已经买过了
              key对应的value用于标记用户之前已经购买了这个场次，这种商品的几件商品
         */
        MemberTo memberTo = LoginInterceptor.MEMBER_TO_THREAD_LOCAL.get();
        String userFlag = SecKillConstant.USER_SEC_KILL_FLAG_PREFIX + memberTo.getId() + "_" + secKillInfoVo.getSpuId() + "_" + secKillInfoVo.getSkuId();

        // 已购买的数量
        Integer purchasedCount = userPurchaseFlagOps.get(userFlag);
        if (!ObjectUtils.isEmpty(purchasedCount) && purchasedCount + secKillInfoVo.getNumber() > secKillRelationAndSpu.getSeckillSkuRelation().getSeckillLimit()) {
            throw new MallException(StatusCodeEnum.SEC_KILL_PURCHASE_MULTIPLE_EXCEEDS_LIMIT);
        }


        // 3、开始抢购
        String productSemaphoreKey = UploadSecKillSkuServiceImpl.getProductSemaphoreKey(secKillInfoVo.getSpuId(), secKillInfoVo.getSkuId(), secKillInfoVo.getRandomCode());
        RSemaphore semaphore = redissonClient.getSemaphore(productSemaphoreKey);

        boolean acquire = false;
        try {
            // 如果获取到信号量（扣库存，就说明抢购成功）

            acquire = semaphore.tryAcquire(secKillInfoVo.getNumber(), 100, TimeUnit.MILLISECONDS);
//            semaphore.acquire(secKillInfoVo.getNumber()); // 这个会阻塞方法的执行，不符合业务需求，如果获取不到信号量，就立刻失败

            if (!acquire) {

                throw new MallException(StatusCodeEnum.SEC_KILL_GET_SEMAPHORE_FAILED);
            }

            // 4、生成订单号
            String orderSn = IdWorker.getTimeId();

            secKillOrderInfoTo.setOrderSn(orderSn);

            secKillOrderInfoTo.setCount(secKillInfoVo.getNumber());

            secKillOrderInfoTo.setSkuId(secKillInfoVo.getSkuId());

            secKillOrderInfoTo.setSpuId(secKillInfoVo.getSpuId());

            Long memberId = LoginInterceptor.MEMBER_TO_THREAD_LOCAL.get().getId();

            secKillOrderInfoTo.setMemberId(memberId);

            rabbitTemplate.convertAndSend(OrderConstant.ORDER_EVENT_EXCHANGE, OrderConstant.SEC_KILL_CREATE_ORDER_ROUTING_KEY, secKillOrderInfoTo);

            // 2.5 设置该用户已经购买了的数量
            userPurchaseFlagOps.set(userFlag, secKillInfoVo.getNumber() + (purchasedCount == null ? 0 : purchasedCount));
            // 设置用户购买量标记位的过期时间为活动结束时间
            redisTemplate.expire(userFlag, secKillRelationAndSpu.getEndTime() - nowTime, TimeUnit.MILLISECONDS);


            return orderSn;

        } catch (InterruptedException e) {
            // 信号量为0，商品已经被抢光啦

            throw new MallException(StatusCodeEnum.SEC_KILL_GET_SEMAPHORE_FAILED);

        } catch (AmqpException e) {

            // 如果发送到消息队列失败，就直接回滚信号量，没有设计重试发送消息的策略

            log.error("{}，现释放商品 spuId = {}的信号量（库存）【{}】件", StatusCodeEnum.SENT_ORDER_INFO_TO_MESSAGE_QUEUE_FAILED.getMsg(), secKillInfoVo.getSpuId(), secKillInfoVo.getNumber());

            semaphore.release(secKillInfoVo.getNumber());

            throw new MallException(StatusCodeEnum.SENT_ORDER_INFO_TO_MESSAGE_QUEUE_FAILED);
        }


    }
}
